package com.bfxy.disruptor.common.consume;

import java.util.concurrent.atomic.AtomicBoolean;

import com.bfxy.disruptor.common.RingBuffer;
import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.producer.Productor;

public final class NoOpEventConsumer implements EventConsumer {
	private final SequencerFollowingSequence sequence;
	private final AtomicBoolean running = new AtomicBoolean(false);

	public NoOpEventConsumer(final RingBuffer<?> sequencer) {
		sequence = new SequencerFollowingSequence(sequencer);
	}

	@Override
	public Sequence getSequence() {
		return sequence;
	}

	@Override
	public void halt() {
		running.set(false);
	}

	@Override
	public boolean isRunning() {
		return running.get();
	}

	@Override
	public void run() {
		if (!running.compareAndSet(false, true)) {
			throw new IllegalStateException("Thread is already running");
		}
	}

	private static final class SequencerFollowingSequence extends Sequence {
		private final RingBuffer<?> sequencer;

		private SequencerFollowingSequence(final RingBuffer<?> sequencer) {
			super(Productor.INITIAL_CURSOR_VALUE);
			this.sequencer = sequencer;
		}

		@Override
		public long get() {
			return sequencer.getCursor();
		}
	}
}
